package com.surfilter.massdata.spark.task;

import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.TreeMap;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.spark.Accumulator;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.VoidFunction;
import org.apache.spark.broadcast.Broadcast;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SQLContext;

import com.act.sparkanalyz.service.impl.SparkService.OutQueueEntity;
import com.act.sparkanalyz.task.ISparkTask;
import com.surfilter.massdata.spark.bean.SAN011;
import com.surfilter.massdata.spark.util.CommonUtils;
import com.surfilter.massdata.spark.util.DateUtil;

/**
 * 访问全球网站情况
 * 本省网民访问全球的网站情况（本省网民是指CIP在BR2002中存在）
 */
public class WebSiteVisitsGlobalTask implements ISparkTask{
	
	private static final Log log = LogFactory.getLog(WebSiteVisitsGlobalTask.class);
	private static final long serialVersionUID = 1L;
	private String outName;
	
	@Override
	public List<OutQueueEntity> execute(Map<String, DataFrame> dataFrames,Map<String, String> commandMap) {
		log.info("==============WebSiteVisitsGlobalTask begin====================");
		List<OutQueueEntity> outList = new ArrayList<OutQueueEntity>();
		try{
			DataFrame cipFrame = dataFrames.get("cip_frame");
			JavaRDD<Row> cipRDD = cipFrame.toJavaRDD();
			SQLContext sqlContext = cipFrame.sqlContext();
			JavaSparkContext sc = new JavaSparkContext(sqlContext.sparkContext());
			
			//本省网民访问全球的网站总量
			long cipCount = getCipFilterCount(cipRDD, sc);
			
			if(cipCount == 0){
				List<SAN011> list = new ArrayList<SAN011>();
		    	JavaRDD<SAN011> javaRDD = sc.parallelize(list);
		    	DataFrame result = sqlContext.createDataFrame(javaRDD, SAN011.class);
		    	outList.add(new OutQueueEntity(outName,result));
		    	return outList;
			}
			
			//省内访问总量
			long dipCount = getDipFilterCount(cipRDD, sc);
			
	        String dayStr = commandMap.get("-d");
	        Date date = DateUtil.getExecDate(DateUtil.getYesterday(),dayStr);
			
			JavaRDD<SAN011> javaRDD = createRdd(sc, cipCount,dipCount, date);
	    	DataFrame df = sqlContext.createDataFrame(javaRDD, SAN011.class);
	    	
			outList.add(new OutQueueEntity(outName, df));
			
			CommonUtils.deleteTaskTableData("SAN011", date, "WEBSITE_COUNT_SUM", 0, "day");
		}catch(Exception e){
			if(log.isErrorEnabled()){
				log.error(e.getMessage(),e);
			}
		}
		return outList;
	}

	@SuppressWarnings("serial")
	private long getDipFilterCount(JavaRDD<Row> cipRDD, JavaSparkContext sc) {
		final Broadcast<TreeMap<Long,Long>> ipBroadcast = sc.broadcast(CommonUtils.getIpExtensionMap());
		final Accumulator<Double> dipAccum = sc.accumulator(0D);
		cipRDD.filter(new Function<Row, Boolean>() {

			@Override
			public Boolean call(Row row){
				try{
					String dip = row.getAs("dip");
					if(CommonUtils.isExists(ipBroadcast.value(),dip)){
						return true;
					}
				}catch(Exception e){
					if(log.isErrorEnabled()){
						log.error(e.getMessage(),e);
					}
				}
				return false;
			}
		}).foreach(new VoidFunction<Row>() {

			@Override
			public void call(Row row) throws Exception {
				dipAccum.add(Double.parseDouble(row.getAs("dns_visit_count").toString()));
			}
		});/*.foreachPartition(new VoidFunction<Iterator<Row>>() {

			@Override
			public void call(Iterator<Row> it) throws Exception {
				while(it.hasNext()){
					Row row = it.next();
					dipAccum.add(Integer.parseInt(row.getAs("dns_visit_count").toString()));
				}
			}
		});*/
		
		//省内访问总量
		long dipCount = new Double(dipAccum.value()).longValue();
		return dipCount;
	}

	@SuppressWarnings("serial")
	private long getCipFilterCount(JavaRDD<Row> cipRDD, JavaSparkContext sc) {
		final Accumulator<Double> cipAccum = sc.accumulator(0D);
		cipRDD.foreach(new VoidFunction<Row>() {

			@Override
			public void call(Row row) throws Exception {
				cipAccum.add(Double.parseDouble(row.getAs("dns_visit_count").toString()));
			}
		});
		/*cipRDD.foreachPartition(new VoidFunction<Iterator<Row>>() {
			
			@Override
			public void call(Iterator<Row> it) throws Exception {
				while(it.hasNext()){
					Row row = it.next();
					cipAccum.add(Integer.parseInt(row.getAs("dns_visit_count").toString()));
				}
			}
		});*/
		
		//本省网民访问全球的网站总量
		long cipCount = new Double(cipAccum.value()).longValue();
		return cipCount;
	}

	private JavaRDD<SAN011> createRdd(JavaSparkContext sc,long cipCount,long dipCount, Date date) {
		List<SAN011> list = new ArrayList<SAN011>();
		
		//省内访问量
    	SAN011 sa = getSAN011(date,"1",dipCount);
    	list.add(sa);
    	
    	//省外访问总量
		long dipOutCounts = cipCount - dipCount;
		SAN011 dipSa = getSAN011(date,"2",dipOutCounts);
		list.add(dipSa);
		
    	JavaRDD<SAN011> javaRDD = sc.parallelize(list);
		return javaRDD;
	}
	
	private SAN011 getSAN011(Date date,String bussValue,long countValue){
		SAN011 sa = new SAN011();
		sa.setYear(DateUtil.getCurrentYear(date));
		sa.setHalf_year(DateUtil.getHalfYear(date));
		sa.setQuarter(DateUtil.getQuarter(date));
		sa.setMonth(DateUtil.getCurrentMonth(date));
		sa.setWeek(DateUtil.getCurrentWeek(date));
		sa.setDay(DateUtil.getCurrentDay(date));
		sa.setBuss_type("WEBSITE_COUNT_SUM");
		sa.setBuss_value(bussValue);
		sa.setCount_value(countValue);
		sa.setSta_range(0);
		return sa;
	}
}
